package me.shiqiang.im.linsteners;

import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.annotation.OnEvent;
import me.shiqiang.im.enums.MessageCmd;
import me.shiqiang.im.msg.MsgBody;
import me.shiqiang.im.msg.SyncMsg;
import me.shiqiang.im.msg.SyncReq;
import me.shiqiang.im.redis.RedisKey;
import me.shiqiang.im.utils.JsonUtils;
import me.shiqiang.im.util.RedisPipelineProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author cuishiqiang
 * @date 2017/12/14 下午5:42
 */
@Component("SyncReqListener")
public class SyncMsgReqListener extends Listener {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    ConcurrentLinkedQueue<RedisPipelineProcessor> queue;

    @Autowired
    StringRedisTemplate redisTemplate;

    @OnEvent(MessageCmd.SYNC_MSG_REQ)
    public void onSyncReq(SocketIOClient client, SyncReq msg) throws IOException {
        Set<String> range = redisTemplate.opsForZSet()
                .rangeByScore(RedisKey.clientMsgQueue(msg.getUserId().toString()), msg.getSyncKey() + 1, Long.MAX_VALUE, 0, 100);
        SyncMsg syncMsg = new SyncMsg(range.size());
        for (String s : range) {
            try {
                MsgBody body = JsonUtils.getInstance().fromJson(s, MsgBody.class);
                syncMsg.getMsgs().add(body);
            } catch (Exception e) {
                redisTemplate.opsForZSet().remove(RedisKey.clientMsgQueue(msg.getUserId().toString()), s.getBytes());

            }
        }
        client.sendEvent(MessageCmd.SYNC_MSG_RES, syncMsg);
//        byte[][] list = range.stream().map(String::getBytes).collect(Collectors.toList()).toArray(new byte[][]{});

        queue.add(connection -> {
            for (String value : range) {
                connection.zRem(RedisKey.clientMsgQueue(msg.getUserId().toString()).getBytes(), value.getBytes());
            }
        });
    }
}
